package com.ibm.cps.processors;

import java.util.ArrayList;
import java.util.Collection;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.sql.expression.function.meta.ValueParam;
import com.ibm.datamodel.timeseries.TimeSeriesSchema;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.RandomUtil;
import com.ibm.util.exception.CPSException;

@Info(Name=MessageFields.INPUT_TYPE,Type=Info.ProcessorType.Other)
public class FullInputProcessor extends BasicProcessor {
	public static final String DATASOURCE_NAME = "DATASOURCE_NAME";
	public static final String TENANT_ID = "TENANT_ID";

	@ValueParam
	private String name;

	private ITimeSeriesSchema parentSchema;
	private ITimeSeriesSchema schema;

	public FullInputProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source, parentSchema);
		if (parentSchema == null) {
			throw new CPSException(ErrorCode.NULL, "Null schema for input " + processorId);
		}
		this.parentSchema = parentSchema;
	}

	@Override
	public void open() throws CPSException {
		this.schema = updateSchemaWithFullQualifiedName(tenantId, name, parentSchema);
	}

	private ITimeSeriesSchema updateSchemaWithFullQualifiedName(String tenantId, String datasourceName, ITimeSeriesSchema parentSchema) throws CPSException{
		ArrayNode dataSourceTags = parentSchema.getTags().deepCopy();
		for (JsonNode jsonNode : dataSourceTags) {
			ObjectNode field = (ObjectNode)jsonNode;
//			ObjectNode fieldWithFullQualifiedName = JSONUtility.newObjectNode();
			field.put(MessageFields.NAME, datasourceName + ":" + field.get(MessageFields.NAME).asText());
//			fieldWithFullQualifiedName.put(MessageFields.DATATYPE, field.get(MessageFields.DATATYPE).asText());
		}
		String schemaid = RandomUtil.getRandomId();
		return new TimeSeriesSchema(tenantId, schemaid, dataSourceTags, 1);
	}

	@Override
	public Collection<ParsedDataMessage> execute(ParsedDataMessage message) throws CPSException {
		Collection<ParsedDataMessage> retMsg = new ArrayList<ParsedDataMessage>();
		retMsg.add(message);
		return retMsg;
	}

	@Override
	public void dispose() {

	}

	@Override
	public ITimeSeriesSchema getTimeSeriesSchema() {
		return schema;
	}

}
